package com.quectel.util.client.redis;

import com.google.common.base.Preconditions;
import com.quectel.util.common.JacksonUtils;
import com.quectel.util.ex.CommonException;
import com.quectel.util.kit.KryoKit;
import lombok.Getter;
import org.apache.commons.lang3.StringUtils;
import org.redisson.Redisson;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;
import org.springframework.boot.autoconfigure.data.redis.RedisProperties;
import org.springframework.data.redis.connection.RedisServerCommands;
import org.springframework.data.redis.connection.ReturnType;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.StringRedisTemplate;

import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.function.Supplier;
import java.util.stream.Collectors;

/**
 * Redis客户端 主要是增强template和Redisson相关功能
 *
 * @author: caolu
 * @email: louis.cao@quectel.com
 * @date: 2023-04-25 18:15:00
 */
public class RedisClient {
    @Getter
    private StringRedisTemplate template;
    @Getter
    private RedissonClient redissonClient;

    /**
     * 无参构造
     */
    private RedisClient() {

    }

    /**
     * 初始化变量
     *
     * @param redisProperties
     * @param template
     */
    public RedisClient(RedisProperties redisProperties, StringRedisTemplate template) {
        this.redissonClient = RedissonClientBuilder.builder().config(redisProperties).build();
        this.template = template;
    }

    /**
     * 队列左边添加一个值
     *
     * @param key
     * @param values
     */
    public void leftPush(String key, String... values) {
        if (key == null || values == null) {
            return;
        }
        for (String value : values) {
            template.opsForList().leftPush(key, value);
        }
    }

    /**
     * 获取list数据
     *
     * @param key
     * @param start
     * @param end
     * @return
     */
    public List range(String key, long start, long end) {
        if (key == null) {
            return null;
        }
        return template.opsForList().range(key, start, end);
    }

    /**
     * 移除redis数据
     *
     * @param key
     * @param count
     * @param value
     */
    public void lRem(String key, long count, String value) {
        if (key == null || count == 0 || value == null) {
            return;
        }
        template.opsForList().remove(key, count, value);
    }

    /**
     * 取出队列所有的值
     *
     * @param key
     * @return
     */
    public List<String> rightPop(String key) {
        if (key == null) {
            return null;
        }
        List<String> result = new ArrayList<>();
        Long size = template.opsForList().size(key);
        if (size == null) {
            return result;
        }
        for (long i = 0L; i < size; i++) {
            result.add(template.opsForList().rightPop(key));
        }
        return result;
    }

    /**
     * ZADD函数
     *
     * @param key
     * @param value 如果一样会覆盖
     * @param score 分数
     */
    public void zAdd(String key, String value, double score) {
        template.opsForZSet().add(key, value, score);
    }

    /**
     * ZADD函数
     *
     * @param key
     * @param value              如果一样会覆盖
     * @param score              分数
     * @param expirationDuration key有效期
     */
    public Boolean zAdd(String key, String value, double score, Duration expirationDuration) {
        String script = "redis.call('ZADD',KEYS[1],ARGV[1],ARGV[2]); " +
                "redis.call('expire',KEYS[1],ARGV[3]);";

        return template.execute((RedisCallback<Boolean>) redisConnection -> redisConnection.eval(
                script.getBytes(StandardCharsets.UTF_8),
                ReturnType.fromJavaType(Boolean.class),
                1,
                key.getBytes(StandardCharsets.UTF_8),
                String.valueOf(score).getBytes(StandardCharsets.UTF_8),
                value.getBytes(StandardCharsets.UTF_8),
                String.valueOf(expirationDuration.getSeconds()).getBytes(StandardCharsets.UTF_8)
        ));
    }


    /**
     * ZRANGE函数
     *
     * @param key
     * @param min
     * @param max
     */
    public Set<String> zRange(String key, double min, double max) {
        return template.opsForZSet().rangeByScore(key, min, max);
    }

    /**
     * ZDEL函数
     *
     * @param key
     * @param values
     * @return 返回的是真正删除value的个数 如果被删除的元素都不存在返回0
     */
    public Long zDel(String key, String... values) {
        return template.opsForZSet().remove(key, Arrays.stream(values).toArray());
    }

    /**
     * 获取滑动窗口指定时间内数量
     *
     * @param key             窗口标识
     * @param periodTimeLimit 滑动窗口时间限制
     * @return
     */
    public Long addItemToSlideWindow(String key, Duration periodTimeLimit) {
        long nowMillis = System.currentTimeMillis();

        // 窗口增加数据
        zAdd(key, String.valueOf(nowMillis), nowMillis, periodTimeLimit.plus(Duration.ofSeconds(1)));

        return getSlideWindowCount(key, periodTimeLimit);
    }

    /**
     * 获取滑动窗口指定时间内数量
     *
     * @param key               窗口标识
     * @param periodTimeLimit   滑动窗口时间限制
     * @return
     */
    public Long getSlideWindowCount(String key, Duration periodTimeLimit) {
        long nowMillis = System.currentTimeMillis();

        // 无用的窗口数据删除掉
        template.opsForZSet().removeRangeByScore(key, 0, nowMillis - periodTimeLimit.toMillis());

        // 查看窗口的数据
        return template.opsForZSet().count(key, 0, nowMillis);
    }

    /**
     * 滚动窗口频率限制
     *
     * @param uniqueKey            业务的唯一标识
     * @param duration             时间段
     * @param times                次数
     * @param incrementAfterExceed true :超过限制后是否继续放入窗口，滑动窗口继续变大 false :不放入窗口等待过期元素消失后继续放入
     *                             场景：true:常用于IP刷新限流 false:常用于短信发送一类
     * @return
     */
    public boolean isExceedLimitTimesAfterIncrement(String uniqueKey, Duration duration, int times, boolean incrementAfterExceed) {
        if (incrementAfterExceed) {
            return addItemToSlideWindow(uniqueKey, duration) > times;
        }
        return getSlideWindowCount(uniqueKey, duration) >= times || addItemToSlideWindow(uniqueKey, duration) > times;
    }

    /**
     * 设置一个key value
     *
     * @param key
     * @param value
     * @return 返回上次key对应的value如果是第一次存入返回null
     */
    public String getSet(String key, String value) {
        if (key == null || value == null) {
            return null;
        }
        return template.opsForValue().getAndSet(key, value);
    }

    /**
     * 设置一个key value
     *
     * @param key
     * @param value
     */
    @Deprecated
    public void set(String key, String value) {
        if (key == null || value == null) {
            return;
        }
        template.opsForValue().set(key, value);
    }

    /**
     * 设置一个key携带有效期
     *
     * @param key
     * @param value
     * @param expirationTime
     * @param timeUnit
     */
    public void set(String key, String value, long expirationTime, TimeUnit timeUnit) {
        if (key == null || value == null) {
            return;
        }
        template.opsForValue().set(key, value, expirationTime, timeUnit);

    }

    /**
     * 得到一个value
     *
     * @param key
     * @return
     */
    public String get(String key) {
        if (key == null) {
            return null;
        }
        return template.opsForValue().get(key);
    }

    /**
     * 设置一个对象携带有效期 其实是把value转为json
     *
     * @param key
     * @param value
     * @param expirationTime
     * @param timeUnit
     */
    public void setObject(String key, Object value, long expirationTime, TimeUnit timeUnit) {
        if (key == null || value == null) {
            return;
        }
        template.opsForValue().set(key, JacksonUtils.toJsonString(value), expirationTime, timeUnit);

    }


    /**
     * 得到对象
     *
     * @param key
     * @param clazz 泛型
     */
    public <T> T getObject(String key, Class<T> clazz) {
        if (key == null || clazz == null) {
            return null;
        }
        String value = get(key);
        if (StringUtils.isBlank(value)) {
            return null;
        }

        return JacksonUtils.parse(value, clazz);
    }

    /**
     * 序列化并存入
     *
     * @param key
     * @param value
     * @param expirationTime
     * @param timeUnit
     */
    public Boolean setBinaryObject(String key, Object value, long expirationTime, TimeUnit timeUnit) {
        if (key == null || value == null) {
            return false;
        }

        return template.execute((RedisCallback<Boolean>) connection -> {
            connection.setEx(
                    key.getBytes(StandardCharsets.UTF_8),
                    timeUnit.toSeconds(expirationTime),
                    KryoKit.serialize(value)
            );
            return true;
        });
    }

    /**
     * 得到序列化对象
     *
     * @param key
     * @return
     */
    public Object getBinaryObject(String key) {
        return template.execute((RedisCallback<Object>) connection -> {
            byte[] objectByteData = connection.get(key.getBytes(StandardCharsets.UTF_8));
            if (objectByteData == null) {
                return null;
            }
            return KryoKit.deserialize(objectByteData);
        });
    }

    /**
     * key s1 zhang
     * key s2 wang
     * key s3 li
     *
     * @param key
     * @param hashKey
     * @param value
     * @param keyExpirationTime 大集合的过期时间
     * @param timeUnit
     */
    public Boolean hSet(String key, String hashKey, String value, long keyExpirationTime, TimeUnit timeUnit) {
        if (key == null || hashKey == null || value == null) {
            return false;
        }
        String script =
                "local current = tonumber(redis.call('EXISTS',KEYS[1])); " +
                        "if (current==0) then " +
                        "redis.call('hset',KEYS[1],ARGV[1],ARGV[2]); " +
                        "redis.call('expire',KEYS[1],ARGV[3]);" +
                        "end;" +
                        "if (current==1) then " +
                        "redis.call('hset',KEYS[1],ARGV[1],ARGV[2]); " +
                        "end;";
        return template.execute((RedisCallback<Boolean>) redisConnection -> redisConnection.eval(
                script.getBytes(StandardCharsets.UTF_8),
                ReturnType.fromJavaType(Boolean.class),
                1,
                key.getBytes(StandardCharsets.UTF_8),
                hashKey.getBytes(StandardCharsets.UTF_8),
                value.getBytes(StandardCharsets.UTF_8),
                String.valueOf(timeUnit.toSeconds(keyExpirationTime)).getBytes(StandardCharsets.UTF_8)
        ));
    }

    /**
     * 通过key和hashKey得到一个值
     *
     * @param key
     * @param hashKey
     * @return
     */
    public String hGet(String key, String hashKey) {
        return template.<String, String>opsForHash().get(key, hashKey);
    }

    /**
     * 得到hset所有的hashKey 和 value
     *
     * @param key
     * @return key代表hashKey
     */
    public Map<String, String> hGetAll(String key) {
        return new LinkedHashMap<>(template.<String, String>opsForHash().entries(key));
    }

    /**
     * 删除hashKey对应的值
     *
     * @param key
     * @param hashKeys
     */
    public Long hDel(String key, Object... hashKeys) {
        if (key == null || hashKeys == null || hashKeys.length <= 0) {
            return 0L;
        }
        return template.<String, String>opsForHash().delete(key, hashKeys);
    }

    /**
     * 判断一个key是否存在
     *
     * @param key
     * @return
     */
    public Boolean exists(String key) {
        if (key == null) {
            return false;
        }
        return template.hasKey(key);
    }

    /**
     * 删除指定的key
     *
     * @param key
     */
    public void del(String key) {
        if (key != null) {
            template.delete(key);
        }
    }

    /**
     * 模糊删除（慎用）
     * 模糊删除 a_* 就代表删除已a_开头的key *a_*删除包含a_的键
     *
     * @param key
     */
    public void delFuzzy(String key) {
        Set<String> keys = template.keys(key);
        if (keys != null && !keys.isEmpty()) {
            template.delete(keys);
        }
    }

    /**
     * 给key设置有效期
     *
     * @param key
     * @param expirationTime
     * @param timeUnit
     */
    public void expire(String key, long expirationTime, TimeUnit timeUnit) {
        template.expire(key, expirationTime, timeUnit);
    }

    /**
     * 得到key剩余的过期时间
     *
     * @param key
     * @param timeUnit
     * @return -2代表key不存在 -1代表key存在但是没有设置有效期
     */
    public Long ttl(String key, TimeUnit timeUnit) {
        return template.getExpire(key, timeUnit);
    }

    /**
     * 自增一个key 默认从0开始自增
     *
     * @param key
     * @param increment 每次自增多少
     * @return 自增后的值
     */
    public Long increment(String key, long increment) {
        return template.opsForValue().increment(key, increment);
    }

    /**
     * 自增设置有效期 已经实现原子性  <br>
     * 对于double的值尽量转为long然后自增 因为redis浮点型自增精度会丢失
     *
     * @param key
     * @param increment
     * @param expirationTime
     * @param timeUnit
     * @return
     */
    public Long increment(String key, long increment, long expirationTime, TimeUnit timeUnit) {
        String script =
                "local current = redis.call('incrby',KEYS[1],ARGV[1]); " +
                        "local t = redis.call('ttl',KEYS[1]); " +
                        "if t ~= -2 then  " +
                        "redis.call('expire',KEYS[1],ARGV[2]) " +
                        "end; " +
                        "return current";
        return template.execute((RedisCallback<Long>) redisConnection -> redisConnection.eval(
                script.getBytes(StandardCharsets.UTF_8),
                ReturnType.fromJavaType(Long.class),
                1,
                key.getBytes(StandardCharsets.UTF_8),
                (String.valueOf(increment)).getBytes(StandardCharsets.UTF_8),
                String.valueOf(timeUnit.toSeconds(expirationTime)).getBytes(StandardCharsets.UTF_8)
        ));
    }

    /**
     * 循环自增设置有效期 已经实现原子性  <br>
     * 对于double的值尽量转为long然后自增 因为redis浮点型自增精度会丢失
     *
     * @param key
     * @param increment      支持负数
     * @param min
     * @param max
     * @param expirationTime
     * @param timeUnit
     * @return
     */
    public Long loopIncrement(String key, long increment, long min, long max, long expirationTime, TimeUnit timeUnit) {

        //KEYS[1]代表key
        //ARGV[1]代表increment
        //ARGV[2]代表min
        //ARGV[3]代表max
        //ARGV[4]代表expirationTime
        String script =
                "local current = tonumber(redis.call('incrby',KEYS[1],ARGV[1])); " +
                        //如果大于最大值则变为最小值
                        "if (current>tonumber(ARGV[3])) then " +
                        "redis.call('set',KEYS[1],ARGV[2]) " +
                        "current =  tonumber(ARGV[2]) " +
                        "end;" +
                        //如果最小则变为最大
                        "if (current< tonumber(ARGV[2])) then " +
                        "redis.call('set',KEYS[1],ARGV[3]) " +
                        "current = tonumber(ARGV[3]) " +
                        "end; " +
                        "redis.call('expire',KEYS[1],ARGV[4]) " +
                        "return current";
        return template.execute((RedisCallback<Long>) redisConnection -> redisConnection.eval(
                script.getBytes(StandardCharsets.UTF_8),
                ReturnType.fromJavaType(Long.class),
                1,
                key.getBytes(StandardCharsets.UTF_8),
                (String.valueOf(increment)).getBytes(StandardCharsets.UTF_8),
                (String.valueOf(min)).getBytes(StandardCharsets.UTF_8),
                (String.valueOf(max)).getBytes(StandardCharsets.UTF_8),
                String.valueOf(timeUnit.toSeconds(expirationTime)).getBytes(StandardCharsets.UTF_8)
        ));
    }

    /**
     * @param key
     * @param hashKey
     * @return
     */
    public Long hIncrement(String key, String hashKey) {
        return hIncrement(key, hashKey, 1);
    }

    /**
     * @param key
     * @param hashKey
     * @param increment
     * @return
     */
    public Long hIncrement(String key, String hashKey, long increment) {
        return template.opsForHash().increment(key, hashKey, increment);
    }

    /**
     * 得到redis时间
     *
     * @return
     */
    public Long getRedisCurrentMillis() {
        return template.execute((RedisCallback<Long>) RedisServerCommands::time);
    }

    /**
     * setnx函数 key存在返回false 不设置有效期慎用
     * set name 1 px 100000 nx
     *
     * @param key
     * @param value
     * @return
     */
    @Deprecated
    public Boolean setNX(String key, String value) {
        return template.opsForValue().setIfAbsent(key, value);
    }

    /**
     * setnx函数 key存在返回false key不存在返回true
     * set name 1 px 100000 nx
     *
     * @param key
     * @param value
     * @param expirationTime
     * @param timeUnit
     * @return
     */
    public Boolean setNX(String key, String value, long expirationTime, TimeUnit timeUnit) {
        return template.opsForValue().setIfAbsent(key, value, expirationTime, timeUnit);
    }

    /**
     * set里边添加值
     *
     * @param key
     * @param values
     */
    public void sAdd(String key, Set<String> values, long expirationTime, TimeUnit timeUnit) {
        if (key == null || values == null) {
            return;
        }
        String script =
                "redis.call('sadd',KEYS[1],ARGV[1]);" +
                        "redis.call('expire',KEYS[1],ARGV[2]);" +
                        "return true;";
        template.executePipelined((RedisCallback<Boolean>) connection -> {
            for (String value : values) {
                connection.eval(
                        script.getBytes(StandardCharsets.UTF_8),
                        ReturnType.fromJavaType(Boolean.class),
                        1,
                        key.getBytes(StandardCharsets.UTF_8),
                        value.getBytes(StandardCharsets.UTF_8),
                        (String.valueOf(timeUnit.toSeconds(expirationTime))).getBytes(StandardCharsets.UTF_8)
                );
            }
            return null;
        });
    }

    /**
     * 取出set的值
     *
     * @param key
     */
    public Set<String> members(String key) {
        if (key == null) {
            return null;
        }
        return template.opsForSet().members(key);
    }

    /*
     * =================== 以下为Redisson客户端的增强 start  ===================
     */

    /**
     * 获取分布式锁
     *
     * @param lockKey
     * @return RedisLock
     */
    public Lock getDistributedLock(String lockKey) {
        return redissonClient.getLock(lockKey);
    }

    /**
     * 快速加锁方法
     *
     * @param lockKey
     * @param supplier
     * @param <T>
     * @return
     */
    public <T> T doInDistributedLock(String lockKey, Supplier<T> supplier) {
        Preconditions.checkArgument(lockKey != null);
        Preconditions.checkArgument(supplier != null);
        Lock spinLock = getDistributedLock(lockKey);
        spinLock.lock();
        try {
            return supplier.get();
        } catch (Exception e) {
            throw new CommonException("Do in redis lock throw exception", e);
        } finally {
            spinLock.unlock();
        }

    }

    /**
     * @param lockKey
     * @return
     */
    public Lock getSpinLock(String lockKey) {
        return redissonClient.getSpinLock(lockKey);
    }

    /**
     * 快速加锁方法
     *
     * @param lockKey
     * @param supplier
     * @param <T>
     * @return
     */
    public <T> T doInSpinLock(String lockKey, Supplier<T> supplier) {
        Preconditions.checkArgument(lockKey != null);
        Preconditions.checkArgument(supplier != null);
        Lock spinLock = getSpinLock(lockKey);
        spinLock.lock();
        try {
            return supplier.get();
        } catch (Exception e) {
            throw new CommonException("Do in redis lock throw exception", e);
        } finally {
            spinLock.unlock();
        }

    }

    /**
     * 同时锁定多个key
     *
     * @param keys
     * @return
     */
    public Lock getMultiSpinLock(String... keys) {
        Preconditions.checkArgument(keys != null && keys.length > 0);
        RLock[] locks = new RLock[keys.length];
        for (int i = 0; i < keys.length; i++) {
            locks[i] = redissonClient.getSpinLock(keys[i]);
        }
        return redissonClient.getMultiLock(locks);
    }

    /*
     * =================== 以下为Redisson客户端的增强 end  ===================
     */


    /**
     * Redisson客户端构造器
     *
     * @author caolu
     */

    public static class RedissonClientBuilder {

        /**
         * Redis配置参数
         */
        private RedisProperties redisProperties;

        /**
         * 无参构造
         */
        private RedissonClientBuilder() {

        }

        public static RedissonClientBuilder builder() {
            return new RedissonClientBuilder();
        }

        public RedissonClientBuilder config(RedisProperties redisProperties) {
            this.redisProperties = redisProperties;
            return this;
        }

        public RedissonClient build() {
            Config config = new Config();
            if (redisProperties.getCluster() != null) {
                // cluster模式
                List<String> newNodes =
                        redisProperties
                                .getCluster()
                                .getNodes()
                                .stream()
                                .map((node) -> (redisProperties.isSsl() ? "rediss://" : "redis://") + node)
                                .collect(Collectors.toList());
                config
                        .useClusterServers()
                        .setClientName(redisProperties.getClientName())
                        .addNodeAddress(newNodes.toArray(new String[]{}))
                        .setPassword(redisProperties.getPassword());

            } else if (redisProperties.getSentinel() != null) {
                // sentinel模式
                RedisProperties.Sentinel sentinel = redisProperties.getSentinel();
                List<String> newNodes =
                        sentinel
                                .getNodes()
                                .stream()
                                .map((node) -> (redisProperties.isSsl() ? "rediss://" : "redis://") + node)
                                .collect(Collectors.toList());
                config
                        .useSentinelServers()
                        .setClientName(redisProperties.getClientName())
                        .setMasterName(sentinel.getMaster())
                        .addSentinelAddress(newNodes.toArray(new String[]{}))
                        .setPassword(redisProperties.getPassword())
                        .setDatabase(redisProperties.getDatabase());
            } else {
                // 单机
                config
                        .useSingleServer()
                        .setClientName(redisProperties.getClientName())
                        .setAddress(String.format("%s://%s:%d", redisProperties.isSsl() ? "rediss" : "redis", redisProperties.getHost(), redisProperties.getPort()))
                        .setPassword(redisProperties.getPassword())
                        .setDatabase(redisProperties.getDatabase());

            }
            return Redisson.create(config);
        }
    }
}
